package com.guagua.util;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.FailoverPolicy;
import me.prettyprint.cassandra.service.template.ColumnFamilyResult;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;

public class CassandraClient {
	
	private static  String keySpace ;// 相当于DB NAME
	private  static  String columnFamily;// 相当于DB Table
	private static String connectionURL;
	private static String username;
	private static String password;
	private static Properties _systemProperties = null;
	private ColumnFamilyTemplate<String, String> template ;
	
	public CassandraClient(){
		if(_systemProperties == null){
			_systemProperties = new Properties();
			try {
				_systemProperties.load(HazelcastConfigUtil.class.getResourceAsStream("/system.properties"));
				keySpace = _systemProperties.getProperty("rankinglist_keyspaceName");
				columnFamily = _systemProperties.getProperty("rankinglist_columnFamily");
				connectionURL = _systemProperties.getProperty("rankinglist_conn_url");
				username = _systemProperties.getProperty("rankinglist_username");
				password = _systemProperties.getProperty("rankinglist_password");
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
		
		
		CassandraHostConfigurator chc = new CassandraHostConfigurator();
		// 把两台server的ip都写上，确保高可用性，有一台写故障时会自动写另一台
		chc.setHosts(connectionURL);
		Cluster cluster = HFactory.getOrCreateCluster("Guagua Data Cluster", chc);
		ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();
		ccl.setDefaultWriteConsistencyLevel(HConsistencyLevel.ONE);
		Map accessMap = new HashMap();
		accessMap.put("username", username);
		accessMap.put("password", password);
		Keyspace ksp = HFactory.createKeyspace(keySpace, cluster, ccl, FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE, accessMap);
		template = new ThriftColumnFamilyTemplate<String, String>(ksp, columnFamily, StringSerializer.get(), StringSerializer.get());
	}
	
	public String getPopularityWeek(String columnKey){
		ColumnFamilyResult<String, String> res = template.queryColumns(columnKey);
		return res.getString("v");
	}
	
	
	
	
	
	
	
	
	
	
	
	
	
	
	
	
	
}